package drds.binlog.sink.group;

import drds.binlog.common.Authors;
import drds.binlog.sink.EntrySink;
import drds.binlog.sink.event_down_stream_handler.EventDownStreamHandler;
import drds.binlog.store.Event;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;

import java.util.Arrays;
import java.util.List;

/**
 * 基于归并排序的sink处理
 * <p>
 * <pre>
 * 几点设计说明：
 * 1. 多库合并时，需要控制不满足groupSize的条件，就会阻塞其他库的合并操作.  (比如刚启动时会所有通道正常工作才开始合并，或者中间过程出现主备切换)
 * 2. 库解析出现问题，但没有进行主备切换，此时需要通过{@linkplain EventDownStreamHandler}进行定时监听合并数据的产生时间间隔
 *    a. 因为一旦库解析异常，就不会再sink数据，此时groupSize就会一直缺少，就会阻塞其他库的合并，也就是不会有数据写入到store中
 * </pre>
 */
@Author(name = Authors.LI_YANG)
public class GroupEntrySink extends EntrySink
{

    @Setter
    @Getter
    private int groupSize;
    @Setter
    @Getter
    private EventMergeGroupBarrier eventMergeGroupBarrier;  // 归并排序需要预先知道组的大小，用于判断是否组内所有的sink都已经开始正常取数据


    public GroupEntrySink(int groupSize)
    {
        super();
        this.groupSize = groupSize;
    }

    public void start()
    {
        super.start();

        if (this.filterTransactionEvent)
        {
            this.eventMergeGroupBarrier = new TimelineEventMergeGroupBarrier(this.groupSize);
        } else
        {
            this.eventMergeGroupBarrier = new TransactionTimelineEventMergeGroupBarrier(this.groupSize);
        }
    }

    public void interrupt()
    {
        super.interrupt();
        eventMergeGroupBarrier.interrupt();
    }

    //
    protected boolean doSink(List<Event> eventList)
    {
        int size = eventList.size();
        for (int i = 0; i < eventList.size(); i++)
        {
            Event event = eventList.get(i);
            try
            {
                eventMergeGroupBarrier.awaitWhenIsNotPermit(event);// 进行timeline的归并调度处理
                if (filterTransactionEvent)
                {
                    return super.doSink(Arrays.asList(event));
                } else if (i == size - 1)
                {
                    // 针对事务数据，只有到最后一条数据都通过后，才进行sink操作，保证原子性
                    // 同时批量sink，也要保证在最后一条数据释放状态之前写出数据，否则就有并发问题
                    return super.doSink(eventList);
                }
            } catch (InterruptedException e)
            {
                return false;
            } finally
            {
                eventMergeGroupBarrier.removeEventExecuteDateTimeLongFromQueue(event);
            }
        }

        return false;
    }


}
